import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.common.functions.JoinFunction;
public class PointWeighter
        implements JoinFunction<Rating, Tuple2<String, Double>, Tuple2<String, Double>> {

    @Override
    public Tuple2<String, Double> join(Rating rating, Tuple2<String, Double> weight) {
        // multiply the points and rating and construct a new output tuple
        return new Tuple2<String, Double>(rating.name, rating.points * weight.f1);
    }
}